package com.duoduo.cannales.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;

/**
 * cannal 配置
 *
 * @author qian
 * @version 1.0
 * @date 2021/3/8 16:10
 */
@Component
public class CanalClient implements DisposableBean {

    private CanalConnector canalConnector;

    @Bean
    public CanalConnector getCanalConnector() {
        canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
                11111), "example", "", "");
        canalConnector.connect();
        //指定filter,格式{database}.{table}
        canalConnector.subscribe(".*\\..*");
        //回滚寻找上次中断的为止
        canalConnector.rollback();
        return canalConnector;
    }


    /**
     * 在spring容器销毁的时候，需要断开canal客户端的连接
     * 防止canal连接的泄露
     *
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        if (canalConnector != null) {
            canalConnector.disconnect();
        }
    }
}
